package net.xuele.learn.flink.book;

import net.xuele.learn.flink.book.utils.SensorReading;
import net.xuele.learn.flink.book.utils.SensorSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.SlidingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

/**
 * @Author patrick
 * @Date 2023/7/7 15:01
 * @Description
 */
public class WindowTest {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.addSource(new SensorSource())
                .keyBy(r -> r.id)
                // 每隔10s就会创建一个长度为1分钟的时间窗口
                .window(SlidingProcessingTimeWindows.of(Time.minutes(1), Time.seconds(10)))
                .process(new SimpleProcessWindowFunction());

        env.execute();
    }

    private static class SimpleProcessWindowFunction extends ProcessWindowFunction<SensorReading, SensorReading, String, TimeWindow> {


        private static final long serialVersionUID = -2548360471468211728L;

        @Override
        public void process(String s, ProcessWindowFunction<SensorReading, SensorReading, String, TimeWindow>.Context context, Iterable<SensorReading> elements, Collector<SensorReading> out) throws Exception {
            TimeWindow window = context.window();
            long start = window.getStart();
            long cnt = 0;
            for (SensorReading element : elements) {
                cnt++;
            }
            System.out.println("开始时间为：" + start + "， 结束时间为：" + window.getEnd() + "， key：" + s + "，cnt：" + cnt);
        }
    }

    private static class TestProcessFunction extends ProcessFunction<String, String> {

        @Override
        public void processElement(String value, ProcessFunction<String, String>.Context ctx, Collector<String> out) throws Exception {

        }
    }
}
